package com.chengqs.dfs.backupnode.server;

import lombok.extern.slf4j.Slf4j;

import java.io.File;
import java.io.FileOutputStream;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;

/**
 * <B>主类名称：</B>FSImageCheckpointer<BR>
 * <B>概要说明：</B>fsimage文件的checkpoint组件<BR>
 *
 * @author: chengqs
 * @create: 2025-04-03 18:30:12
 **/
@Slf4j
public class FSImageCheckpointer extends Thread {

    /**
     * checkpoint操作的间隔
     */
    public static final Integer CHECKPOINT_INTERVAL = 60 * 1000;

    private BackupNode backupNode;

    private FSNameSystem nameSystem;

    private NameNodeRpcClient nameNode;

    private String lastFSImageFile = "";

    private long checkpointTime = -1;

    public FSImageCheckpointer(BackupNode backupNode, FSNameSystem nameSystem, NameNodeRpcClient nameNode) {
        this.backupNode = backupNode;
        this.nameSystem = nameSystem;
        this.nameNode = nameNode;
        this.checkpointTime = System.currentTimeMillis();
    }

    @Override
    public void run() {
        log.info("fsimage checkpoint定时调度线程启动......");
        while (backupNode.isRunning()) {
            try {
                if(!nameSystem.isFinishedRecover()) {
                    log.info("当前还没完成元数据恢复，不进行checkpoint......");
                    Thread.sleep(1000);
                    continue;
                }

                if(lastFSImageFile.equals("")) {
                    this.lastFSImageFile = nameSystem.getCheckpointFile();
                }

                long now = System.currentTimeMillis();

                if (now - this.checkpointTime > CHECKPOINT_INTERVAL) {
                    if (!nameNode.isNameNodeRunning()) {
                        log.info("namenode当前无法访问，不执行checkpoint......");
                        continue;
                    }

                    // 一小时之后，进行fsimage的checkpoint操作，去把内存中的数据写入磁盘就可以了
                    // 写数据的过程中，必须是
                    log.info("准备执行checkpoint操作，写入fsimage文件......");
                    doCheckpoint();
                    log.info("完成checkpoint操作，写入fsImage文件......");
                }

                Thread.sleep(1000);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 将fsImage持久化到磁盘
     * @throws Exception
     */
    private void doCheckpoint() throws Exception {
        FSImage fsImage = nameSystem.getFSImage();
        removeLastFSImageFile();
        writeFSImageFile(fsImage);
        uploadFSImageFile(fsImage);
        updataCheckpointTxid(fsImage);
        saveCheckpontInfo(fsImage);
    }

    /**
     * 持久化checkpoint信息
     * @param fsImage
     */
    private void saveCheckpontInfo(FSImage fsImage) {
        String path = "D:\\java\\儒猿课堂\\儒猿海量小文件分布式存储项目\\backupnode\\checkpoint-info.meta";

        RandomAccessFile raf = null;
        FileOutputStream out = null;
        FileChannel channel = null;

        try {
            File file = new File(path);
            if (file.exists()) {
                file.delete();
            }

            long now = System.currentTimeMillis();
            this.checkpointTime = now;
            long checkpointTxid = fsImage.getMaxTxid();
            ByteBuffer buffer = ByteBuffer.wrap(String.valueOf(now + "_" + checkpointTxid + "_" + lastFSImageFile).getBytes());

            raf = new RandomAccessFile(path, "rw");
            out = new FileOutputStream(raf.getFD());
            channel = out.getChannel();

            channel.write(buffer);
            channel.force(false); // 强制把数据刷到磁盘上

            log.info("checkpoint信息持久化到磁盘文件......");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (out != null) {
                    out.close();
                }
                if (raf != null) {
                    raf.close();
                }
                if (channel != null) {
                    channel.close();
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 更新checkpointTxid
     * @param fsImage
     */
    private void updataCheckpointTxid(FSImage fsImage) {
        nameNode.updateCheckpointTxid(fsImage.getMaxTxid());
    }

    /**
     * 删除上一个fsimage文件
     */
    private void removeLastFSImageFile() throws Exception {
        File file = new File(lastFSImageFile);
        if (file.exists()) {
            file.delete();
        }
    }

    /**
     * 写入最新的FSImage文件
     * @throws Exception
     */
    private void writeFSImageFile(FSImage fsImage) throws Exception {

        ByteBuffer buffer = ByteBuffer.wrap(fsImage.getFsImageJson().getBytes());

        // fs文件名的格式，应该包含了当前这个里面最后一个edits log的txid
        String fsImageFile = "D:\\java\\儒猿课堂\\儒猿海量小文件分布式存储项目\\backupnode\\fsimage-" +
                fsImage.getMaxTxid() + ".meta";
        lastFSImageFile = fsImageFile;

        RandomAccessFile file = null;
        FileOutputStream out = null;
        FileChannel channel = null;

        try {
            file = new RandomAccessFile(fsImageFile, "rw");
            out = new FileOutputStream(file.getFD());
            channel = out.getChannel();

            channel.write(buffer);
            channel.force(false); // 强制把数据刷到磁盘上
        } finally {
            if (out != null) {
                out.close();
            }
            if (file != null) {
                file.close();
            }
            if (channel != null) {
                channel.close();
            }
        }
    }

    /**
     * 上传最新的fsimage文件
     * @param fsImage
     * @throws Exception
     */
    private void uploadFSImageFile(FSImage fsImage) throws Exception {
        FSImageUploader fsImageUploader = new FSImageUploader(fsImage);
        fsImageUploader.start();
    }
}
